package com.lc.dubbo.consumer.filter;

import org.apache.curator.x.discovery.strategies.StickyStrategy;
import org.apache.dubbo.common.constants.CommonConstants;
import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.rpc.*;

import java.sql.Array;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.*;
import java.util.stream.IntStream;
import java.util.stream.Stream;

/**
 * @author liuc
 * @date 2021/6/17
 * @description
 **/
@Activate(group = {CommonConstants.CONSUMER})
public class TPMonitorFilter implements Filter {

    private static ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(1);
    private static ExecutorService executorService = Executors.newSingleThreadExecutor();
    private static Map<String, TPStatisticMap> map = new ConcurrentHashMap<>();
    static {
        executor.scheduleAtFixedRate(new TimeConsumeStatistic(),5,5, TimeUnit.SECONDS);
    }

    @Override
    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
        //记录耗时
        long start = System.currentTimeMillis();
        Result res = invoker.invoke(invocation);
        long end = System.currentTimeMillis();
        executorService.execute(() -> map.putIfAbsent(invocation.getMethodName(),new TPStatisticMap(invocation.getMethodName())).put(new TimeConsumer(start),end-start));
        //System.out.println("invoke time : " + (end-start));
        return res;
    }

    /**
     * 耗时统计
     */
    static class TimeConsumeStatistic extends Thread{

        @Override
        public void run() {
            System.out.println("打印一次最近1分钟内每个方法的TP90、TP99的耗时情况: ");
            //打印一次最近1分钟内每个方法的TP90、TP99的耗时情况
            for (Map.Entry<String, TPStatisticMap> e : map.entrySet()) {
                String methodName = e.getKey();
                TPStatisticMap m = e.getValue();
                Collection<Long> values = m.values();
                if(values == null || values.size() == 0){
                    continue;
                }
                int[] arr = new int[values.size()];
                int i = 0;
                for (Long t : values) {
                    if(t == null || t == 0){
                        continue;
                    }
                    arr[i++] = t.intValue();
                }
                Arrays.sort(arr);
                System.out.println("method ["+methodName+"] statistic: invoke times of latest one minute: ["+arr.length+"], tp50: ["+arr[arr.length/2]+"], tp99: ["+arr[(int)(arr.length*0.99)]+"]");
            }
        }
    }

    /**
     * 记录调用耗时
     */
    static class TimeConsumer{
        //调用时间
        private long invokeTime;

        private static final SimpleDateFormat SIMPLE_DATE_FORMAT = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

        @Override
        public String toString() {
            return "TimeConsumer{" +
                    "invokeTime=" + SIMPLE_DATE_FORMAT.format(new Date(invokeTime)) +
                    '}';
        }

        public TimeConsumer(long invokeTime) {
            this.invokeTime = invokeTime;
        }
    }

    /**
     * 用于存储调用耗时记录，key: 调用时间 value: 耗时(毫秒)
     */
    static class TPStatisticMap extends ConcurrentHashMap<TimeConsumer,Long>{
        //一分钟毫秒数
        private static final long ONE_MINUTE = 60000;
        
        private boolean running = false;

        private LinkedList<TimeConsumer> keyList = new LinkedList<>();

        private String methodName;

        private ScheduledThreadPoolExecutor SCHEDULED_THREAD_POOL_EXECUTOR = new ScheduledThreadPoolExecutor(1);

        public TPStatisticMap(String methodName) {
            this.methodName = methodName;
        }

        @Override
        public Long put(TimeConsumer key, Long value) {
            if(!running){
                running = true;
                SCHEDULED_THREAD_POOL_EXECUTOR.scheduleAtFixedRate(() ->{
                    //System.out.println(methodName+", key size : " + keyList.size());
                    Iterator<TimeConsumer> iterator = keyList.iterator();
                    long nowMills = System.currentTimeMillis(); 
                    while (iterator.hasNext()){
                        TimeConsumer next = iterator.next();
                        //将一分钟之前的耗时调用记录删除
                        if((nowMills - next.invokeTime) > ONE_MINUTE){
                            iterator.remove();
                            this.remove(next);
                        }else{
                            //key是用linkedList存储的，所以可以直接break，后面的key肯定是更近的
                            break;
                        }
                    }
                },5,5,TimeUnit.SECONDS);
            }
            keyList.add(key);
            return super.put(key, value);
        }

        /*@Override
        protected boolean removeEldestEntry(Map.Entry<TimeConsumer, Long> eldest) {
            //一分钟之前的需要删除，不需要被统计
            long nowMills = System.currentTimeMillis();
            return nowMills - eldest.getKey().invokeTime > ONE_MINUTE;
        }*/
    }

}
